package org.example.flinktest;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.example.flinktest.operators.LineSplitter;

/**
 * @author SHB
 */
public class WcStreamDemo {

  public static void main(String[] args) throws Exception {
    //读取参数
    ParameterTool params = ParameterTool.fromArgs(args);
    String host =params.get("host");
    int port=params.getInt("port");


    //定义流环境
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    //检查点设置
    env.enableCheckpointing(6000);
    //设置检查点模式
    env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);
    //checkpoint失败是否job失败
    env.getCheckpointConfig().setFailOnCheckpointingErrors(false);
    //同时运行checkpoint的最大值
    env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
    //
    env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
    //重启策略
    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,500));


    //接收socket流
    DataStream<String> source = env.socketTextStream(host,port);

    /*任务链可以将并行度相同，算子是一对一传输进行整合（redistributing会断开）
    * 减少传输,序列化和反序列化的开销。
    * 对于任务链可以自定义控制：
    * 1.全局禁用：所有算子都打散
    * 2.局部禁用：禁用的算子不会进行任务链
    * 3.主动创建：算子会开始新的任务链
    */
    env.disableOperatorChaining();//1.全局禁用

    //逐一读取数据，打散之后运行wordcount
    DataStream<Tuple2<String, Integer>> result = source
        .flatMap(new LineSplitter())
        .disableChaining()//2.局部禁用
        .startNewChain()//3.主动创建
        .keyBy(0)
        .sum(1);

    //打印输出
    result.print().setParallelism(2);
    //执行任务
    env.execute("Stream word count job");

  }

}
